package com.xzzz.irda.registry.discovery.reactive;

import java.util.function.Function;

import com.xzzz.irda.registry.discovery.RedisDiscoveryClient;
import lombok.extern.slf4j.Slf4j;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import org.springframework.cloud.client.ServiceInstance;
import org.springframework.cloud.client.discovery.ReactiveDiscoveryClient;

@Slf4j
public class RedisReactiveDiscoveryClient implements ReactiveDiscoveryClient {


	private RedisDiscoveryClient discoveryClient;

	public RedisReactiveDiscoveryClient(RedisDiscoveryClient discoveryClient) {
		this.discoveryClient = discoveryClient;
	}

	@Override
	public String description() {
		return "Spring Cloud Redis Reactive Discovery Client";
	}

	@Override
	public Flux<ServiceInstance> getInstances(String serviceId) {

		return Mono.justOrEmpty(serviceId).flatMapMany(loadInstancesFromNacos())
				.subscribeOn(Schedulers.boundedElastic());
	}

	private Function<String, Publisher<ServiceInstance>> loadInstancesFromNacos() {
		return serviceId -> Mono.justOrEmpty(discoveryClient.getInstances(serviceId))
				.flatMapMany(Flux::fromIterable);
	}

	@Override
	public Flux<String> getServices() {
		return Flux.defer(() -> Mono.justOrEmpty(discoveryClient.getServices())
				.flatMapMany(Flux::fromIterable)).subscribeOn(Schedulers.boundedElastic());
	}

}
